package cn.me.alphamq.nameserver.task.cluster.masterslave;

import cn.me.alphamq.common.entity.SlaveAckEntity;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.sync.StopWatch;
import cn.me.alphamq.nameserver.cache.Container;
import cn.me.alphamq.nameserver.enumeration.ReplicationMsgCodeEnum;
import cn.me.alphamq.nameserver.enumeration.ReplicationTypeEnum;
import cn.me.alphamq.nameserver.event.model.cluster.ReplicationMsgEvent;
import cn.me.alphamq.nameserver.store.ReplicationMsgQueueManager;
import cn.me.alphamq.common.store.StopWatchManager;
import cn.me.alphamq.nameserver.task.cluster.ReplicationTask;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * 主从同步中主节点数据发送的任务
 *
 * @author f
 */
public class MasterSendReplicationMsgTask extends ReplicationTask {

    private static final Logger log = LoggerFactory.getLogger(MasterSendReplicationMsgTask.class);

    @Override
    protected void replicationTask() {
        log.info("start sending replication msg task...");
        ReplicationMsgQueueManager replicationMsgQueueManager = Container.getReplicationMsgQueueManager();
        ReplicationTypeEnum type = Container.getClusterProperties()
                .getMasterReplicationProperties().getType();

        while (true) {
            try {
                ReplicationMsgEvent replicationMsgEvent = replicationMsgQueueManager.get();

                // 给从节点发送复制消息
                sendReplicationMsgToSlave(new TcpMsg(EventCodeEnum.MASTER_REPLICATION_MSG_RESP,
                        JSON.toJSONBytes(replicationMsgEvent)));

                // 不需要保证 移除服务实例消息 的强一致性，跳过响应操作
                if (ReplicationMsgCodeEnum.DELETE.getCode() == replicationMsgEvent.getMsgCode()) {
                    continue;
                }

                // 针对不同的复制模式，进行不同的响应操作
                String replicationMsgId = replicationMsgEvent.getMsgId();
                StopWatch stopWatch = StopWatchManager.get(replicationMsgId);
                if (ReplicationTypeEnum.ASYNC == type) {
                    // 对于异步复制，给从节点发送完复制消息后，继续执行原先的逻辑
                    stopWatch.pause();
                } else if (ReplicationTypeEnum.SYNC == type) {
                    // 对于同步复制来说，需要接收到的 ack 事件的数量 就是现有的可用 ctx 数量
                    int needAckCount = Container.getReplicationChannelManager().getValidCtxMap().size();
                    Container.getReplicationAckManager()
                            .putSlaveAck(replicationMsgId, new SlaveAckEntity(needAckCount));
                } else if (ReplicationTypeEnum.HALF_SYNC == type) {
                    // 对于半同步复制来说，需要接收到的 ack 事件的数量 就是现有的可用 ctx 数量 / 2
                    int needAckCount = Container.getReplicationChannelManager().getValidCtxMap().size() / 2;
                    Container.getReplicationAckManager()
                            .putSlaveAck(replicationMsgId, new SlaveAckEntity(needAckCount));
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    // 给从节点发送复制消息，顺便移除非可用的从节点的 channel
    private static void sendReplicationMsgToSlave(TcpMsg replicationMsg) {
        Map<String, ChannelHandlerContext> ctxMap = Container.getReplicationChannelManager().getValidCtxMap();
        // 给从节点发送同步消息，同时挑出非可用的 channel
        for (String ctxId : ctxMap.keySet()) {
            ctxMap.get(ctxId).writeAndFlush(replicationMsg);
        }
    }

}
